package cn.xeblog.server.service;

import cn.xeblog.commons.entity.Response;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * 默认的历史消息服务
 *
 * @author anlingyi
 * @date 2021/9/11 5:10 下午
 */
public class DefaultResponseHistoryService extends AbstractResponseHistoryService {

  private final ExecutorService threadPool = new ThreadPoolExecutor(1, 2, 60,
      TimeUnit.SECONDS, new LinkedBlockingQueue<>(1024), new ThreadPoolExecutor.CallerRunsPolicy());

  /**
   * 最近历史消息的条数
   */
  private final int size;
  /**
   * 历史消息数组
   */
  private final Response[] queue;
  /**
   * 读写锁
   */
  private final ReentrantReadWriteLock readWriteLock;
  /**
   * 消息的数量
   */
  private int count;
  /**
   * 当前从队列获取消息的下标
   */
  private int takeIndex;

  /**
   * 消息存放至队列的下标
   */
  private int putIndex;

  /**
   * 默认的历史消息服务
   */
  public DefaultResponseHistoryService() {
    // 默认只在内存里存最近50条
    this(50);
  }

  /**
   * 指定历史消息条数的历史消息服务
   *
   * @param size 最近历史消息的条数
   */
  public DefaultResponseHistoryService(int size) {
    this.size = size;
    this.queue = new Response[size];
    // 公平锁
    this.readWriteLock = new ReentrantReadWriteLock(true);
  }

  @Override
  protected void addHistoryHandler(Response response) {
    threadPool.execute(() -> offer(response));
  }

  /**
   * 插入历史消息
   *
   * @param response 历史消息
   * @return true, 插入成功，否则，false
   */
  private boolean offer(Response response) {
    // 加写锁
    readWriteLock.writeLock().lock();
    try {
      // 如果满了，移除队列的头
      if (count == size) {
        poll();
      }

      // 放置下标
      queue[putIndex] = response;
      count++;

      // 超出容量从头开始
      if (++putIndex == size) {
        putIndex = 0;
      }
      return true;
    } finally {
      // 解写锁
      readWriteLock.writeLock().unlock();
    }
  }

  /**
   * 获取并移除此队列的头，如果此队列为空，则返回 null
   *
   * @return 队列的头，如果此队列为空，则返回 null
   */
  private Response poll() {

    if (count == 0) {
      return null;
    }

    //获取指定下标的消息，并将该位置置空，即移除指定下标的元素
    Response obj = queue[takeIndex];
    queue[takeIndex] = null;
    count--;

    // 超出容量从头开始
    if (++takeIndex == size) {
      takeIndex = 0;
    }

    return obj;
  }

  @Override
  public List<Response> getHistory(int limit) {
    List<Response> list = new ArrayList<>(limit);

    readWriteLock.readLock().lock();
    // 找最近的limit条消息，
    try {
      // 读取的数量
      int readCount = Math.min(limit, count);
      // 开始的下标
      int start = takeIndex;
      // 超出的长度
      int outLen = start - readCount;

      // 没有超出长度，想当于回退了readCount个下标，当前下标满足回退下标的条件
      if (outLen >= 0) {
        start = outLen;
      } else {
        // 超出了长度，这里是先把回退下标，
        start -= readCount;

        // 如果回退下标后下标为负数，说明回退超出了范围，向前追加
        if (start < 0) {
          start += count;
        }
      }

      // 前面已经把start回退到最近的limit条的起始下标，这里添加最近的limit条数据
      for (int i = 0; i < readCount; i++) {
        list.add(queue[start]);
        // 超出容量从头开始
        if (++start == size) {
          start = 0;
        }
      }

      return list;
    } finally {
      readWriteLock.readLock().unlock();
    }
  }

}
